package net.bw.realtime.jtp.dim.trade.function;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import org.apache.flink.api.common.functions.RichFilterFunction;
import org.apache.flink.configuration.Configuration;

import java.util.HashSet;
import java.util.Set;

/*
 * @ Author：liuyawei
 * @ Date：2025-06-02
 */
public class DimDataFilterFunction extends RichFilterFunction<String> {

    // 获取维度表名
    private Set<String> hashSet;

    // 操作类型
    private Set<String> operatorTypeSet = new HashSet<String>(){{
        add("insert");  add("update"); add("delete");
    }};

    // 构造函数
    public DimDataFilterFunction(Set<String> hashSet) {
        this.hashSet = hashSet;
    }

    //  初始化
    @Override
    public void open(Configuration parameters) throws Exception {
        super.open(parameters);
    }

    //  过滤
    @Override
    public boolean filter(String value) throws Exception {
        // 解析json
        JSONObject jsonObject = JSON.parseObject(value);
        // 获取表名
        String tableName = jsonObject.getString("table_name");
        // 获取操作类型
        String operatorType = jsonObject.getString("operator_type");
        // 判断
        return hashSet.contains(tableName) && operatorTypeSet.contains(operatorType);
    }

    //  释放资源
    @Override
    public void close() throws Exception {
        super.close();
    }
}
